package com.monkeyboy.demo.producer;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.Future;

/**
 * @Author Gavin
 * @date 2020.08.24 09:53
 */
public class SampleProducer {
    public final static String TOPIC_NAME = "first";

    public static void main(String[] args) throws Exception {
        producerSendAsync();
    }
    /*
    Producer异步发送演示
     */
    public static void producerSendAsync() throws Exception {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.150:9092");
        properties.put(ProducerConfig.ACKS_CONFIG, "all");
        properties.put(ProducerConfig.RETRIES_CONFIG, "0");

        //数据达到16k会发送
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, "16384");
        //当时间到达1ms的时候也会发送和上面两个结合起来使用
        properties.put(ProducerConfig.LINGER_MS_CONFIG, "1");
        //数据缓冲区，32M
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432");

        //序列化设置
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        //Producer的主对象
        Producer<String, String> producer = new KafkaProducer<>(properties);
        //组织消息对象：ProducerRecord
        for (int i = 0; i < 100; i++) {

            ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC_NAME, "key:" + i, "value:" + i);

            producer.send(record);
        }
        Thread.sleep(1000);
        //如果这里关闭的时候上面执行的时间没有达到1ms，那么消息会被清空，然后消息也不会进行发送
        producer.close();
    }
    /*
      Producer异步阻塞发送--同步发送
       */
    public static void producerSendSync() throws Exception {
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.1.150:9092");
        properties.put(ProducerConfig.ACKS_CONFIG, "all");
        properties.put(ProducerConfig.RETRIES_CONFIG, "0");
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, "16384");
        properties.put(ProducerConfig.LINGER_MS_CONFIG, "1");
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        //Producer的主对象
        Producer<String, String> producer = new KafkaProducer<>(properties);
        //组织消息对象：ProducerRecord
        for (int i = 0; i < 10; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>(TOPIC_NAME, "key:" + i, "value:" + i);
            Future<RecordMetadata> send = producer.send(record);
            RecordMetadata recordMetadata = send.get();//get方法会阻塞，这就是和下面异步发送的最主要的区别
            System.out.println("partition: " + recordMetadata.partition() + ",offset: " + recordMetadata.offset());
        }
        producer.close();
    }
}
